package com.zzz.OutputFormatOwn;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class MyOutPutFormat extends FileOutputFormat<Text, NullWritable> {
    @Override
    public RecordWriter getRecordWriter(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        FileSystem fileSystem = FileSystem.get(taskAttemptContext.getConfiguration());

        Path goodComment = new Path("file:///E:\\大数据\\数据\\goodComment\\goodComment.txt");
        Path badComment = new Path("file:///E:\\大数据\\数据\\badComment\\badComment.txt");

        FSDataOutputStream goodOutputStream = fileSystem.create(goodComment);
        FSDataOutputStream badOutputStream = fileSystem.create(badComment);

        return new MyRecordWriter(goodOutputStream,badOutputStream);
    }


    public static class MyRecordWriter extends RecordWriter<Text, NullWritable> {
        FSDataOutputStream goodStream = null;
        FSDataOutputStream badStream = null;

        public MyRecordWriter(FSDataOutputStream goodOutputStream, FSDataOutputStream badOutputStream) {
            this.goodStream = goodOutputStream;
            this.badStream = badOutputStream;
        }

        @Override
        public void write(Text text, NullWritable nullWritable) throws IOException, InterruptedException {
            if (text.toString().split("\t")[9].equals("0")){ //注意这里的0是string类型
                goodStream.write(text.toString().getBytes());
                goodStream.write("\r\n".getBytes());
            }else{
                badStream.write(text.toString().getBytes());
                badStream.write("\r\n".getBytes());
            }
        }

        @Override
        public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
            if (goodStream != null){
                goodStream.close();
            }
            if (badStream != null){
                badStream.close();
            }
        }
    }
}
